package com.supermarket.kafka;

import com.supermarket.kafka.generator.DataGenerator;
import com.supermarket.kafka.generator.ProductRecord;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

//作用：将生成的商品记录发送到 Kafka 消息队列
//关键配置：
//可靠性配置：
//acks=all：所有 ISR 副本确认后才认为发送成功
//retries=5：失败自动重试 5 次
//enable.idempotence=true：开启幂等性，防止重复消息
//性能配置：
//未显式设置 batch.size 和 linger.ms，使用默认值
//发送模式：
//异步发送带回调，确保消息发送状态可知
//定时任务：每 5 秒生成一条商品记录

public class KafkaProducerMain {
    private static final String TOPIC = "supermarket00";
    private static final String BOOTSTRAP_SERVERS = "niit01:9092";


    public static void main(String[] args) {
        // 1. 配置Kafka生产者
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 2. 生产级配置(保证数据可靠性)  提高吞吐量和可靠性配置（生产级）
        props.put(ProducerConfig.ACKS_CONFIG, "all");            // 所有副本写入成功才认为写入成功.
        props.put(ProducerConfig.RETRIES_CONFIG, 5);             // 启用重试机制
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1000); // 重试间隔1s
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 幂等性开启，防止重复消息

        // 3. 创建生产者
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 4. 定时任务(每5秒生成一条记录)
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(() -> {
            ProductRecord record = DataGenerator.generateRealisticRecord();
            String value = record.toString();

            ProducerRecord<String, String> kafkaRecord = new ProducerRecord<>(TOPIC, value);
            // 5.异步发送 + 回调函数，实时监控发送状态
            producer.send(kafkaRecord, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("Failed to send message: " + exception.getMessage());
                } else {
                    System.out.printf("Sent message to topic %s, partition %d, offset %d%n",
                            metadata.topic(), metadata.partition(), metadata.offset());
                }
            });
        }, 0, 5, TimeUnit.SECONDS);
    }
}